package com.ourhours.redismq;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskRejectedException;

/**
 * 消息监听器
 *
 * @author dingxiangyong 2016年7月28日 下午2:12:08
 */
public abstract class AbstarctMessageHandler implements Runnable {

	private static Logger logger = LoggerFactory.getLogger(AbstarctMessageHandler.class);

	@Autowired
	RedisUtil redisUtil;
	@Autowired
	RedisMQMessageSender messageSender;

	/**
	 * 初始化启动本监听器
	 */
	@PostConstruct
	public void startListen() {
		// 启动监听
		new Thread(this).start();
	}

	/**
	 * 监听的消息类型
	 */
	private String queueName;

	private boolean monitor;
	
	/**
	 * 是否单线程
	 */
	private boolean single=false;

	// 默认为3
	private int retryTimes = 3;
	
	public AbstarctMessageHandler(String queueName, int retryTimes) {
		this.queueName = MtConstant.KEY_PREFIX + queueName;
		this.retryTimes = retryTimes;
	}
	
	public AbstarctMessageHandler(String queueName, int retryTimes,boolean single) {
		this.queueName = MtConstant.KEY_PREFIX + queueName;
		this.retryTimes = retryTimes;
		this.single = single;
	}

	public AbstarctMessageHandler(String queueName) {
		this.queueName = MtConstant.KEY_PREFIX + queueName;
	}

	public void run() {
		logger.info("----启动redis消息监听：-------------queue：" + this.queueName);
		// 添加到队列名称集合
		redisUtil.sadd(MtConstant.QUEUE_NAME_SET, queueName);
		while (true) {
			try {
				listen();
			} catch (Exception e) {
				logger.error(e.getMessage(), e);
			}
		}
	}

	/**
	 * 监听消息
	 * 
	 * @author lizongyue
	 * @date 2018-02-07
	 */
	@SuppressWarnings("unchecked")
	public void listen() {
		final String queue = queueName;
		logger.debug("=========blpop读取队列前："+queueName);
		// 阻塞获取redis列表值
		final Message message = redisUtil.blpop(queue, Integer.MAX_VALUE, Message.class);
		
		logger.debug("=========blpop读取到消息："+message);

		// 如果获取失败,以为着redis连接有问题，或者消息有问题
		if (message == null) {
			monitor = false;
			logger.warn("消息分发器获取redis连接失败或消息格式非法");
			// 暂停5秒钟,防止线程空跑,打印无数日志
			try {
				Thread.sleep(5000);
			} catch (InterruptedException e) {
				logger.warn("消息分发器线程暂停失败");
			}
			return;
		}
		
		//增加出队数量
		redisUtil.hincrby(MtConstant.QUEUE_OUT_NUMBERS, queue, 1L);
		
		if (!monitor) {
			logger.warn("消息分发开始");
			monitor = true;
		}
		
		if(this.single) {
			handlerMessage(message);
		}else {
			try {
				messageSender.getThreadPool().submit(new Runnable() {
					public void run() {
						handlerMessage(message);
					}
				});
			} catch (TaskRejectedException ex) {
				logger.warn("线程池已满，准备回写任务，暂停本线程");
				// 如果发生任务拒绝加入线程池，则回写任务到redis，等待重试
				messageSender.put(message);

				// 暂停指定时间
				try {
					Thread.sleep(messageSender.getThreadPoolFullSleepSeconds() * 1000);
				} catch (InterruptedException e) {
					logger.warn("生产者暂停异常", ex);
				}
			} catch (Exception ex) {
				logger.error("消息总线发生异常", ex);
			}
		}
	}
	
	/**
	 * 处理消息
	 * @param message
	 */
	void handlerMessage(Message message){
		try {
			// 如果获取成功，则交给子类做业务处理
			handle(message.getContent());
		} catch (Exception ex) {
			logger.error(ex.getMessage(), ex);
			// 处理失败，判断是否需要重试
			if (message.getFailTimes().intValue() >= retryTimes) {
				handleFailed(message.getContent());
			} else {
				message.setFailTimes(message.getFailTimes().intValue() + 1);
				// 再次put回消息总线，等待下次重试
				messageSender.put(message);

				if (logger.isDebugEnabled()) {
					StringBuilder sb = new StringBuilder();
					sb.append("msg:[").append(message).append("], 执行失败，准备重试。");
					logger.debug(sb.toString());
				}
			}
		}
	}

	/**
	 * 获取到消息后做业务处理
	 */
	public abstract void handle(String messageBody);

	/**
	 * 消息多次重试处理失败
	 */
	public abstract void handleFailed(String messageBody);

	public String getMessageType() {
		return queueName;
	}
}
